package com.study.one.pulsar.consumer;

import com.alibaba.fastjson.JSONObject;
import com.study.one.propertise.ConsumerProperties;
import com.study.one.propertise.OneSettingsProperties;
import com.study.one.pulsar.PulsarClientFactory;
import com.study.one.pulsar.messagedto.OrderMessageDTO;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageListener;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;

@Slf4j
@Component
public class PulsarOneOrderConsumer implements ApplicationRunner {

    @Resource
    private OneSettingsProperties oneSettingsProperties;

    @Resource
    private PulsarClientFactory pulsarClientFactory;

    @Override
    public void run(ApplicationArguments args) throws Exception {

        ConsumerProperties.ConsumerTopicProperties oneOrderConsumer = oneSettingsProperties.pulsar.getConsumer().getOneOrderConsumer();
        pulsarClientFactory.createSharedConsumer(oneOrderConsumer.getTopic(), oneOrderConsumer.getSubscription(), new PulsarOneOrderConsumer.OneOrderListener());
    }

    class OneOrderListener implements MessageListener<byte[]> {

        @Override
        @SneakyThrows
        public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
            String message = new String(msg.getValue(), StandardCharsets.UTF_8);
            OrderMessageDTO orderMessageDTO = JSONObject.parseObject(message, OrderMessageDTO.class);
            log.info(orderMessageDTO.toString());
            consumer.acknowledge(msg);
        }
    }


}
